package com.zhuwei.kafka.controller;


import com.google.gson.Gson;
import com.zhuwei.kafka.constant.Topic;
import com.zhuwei.kafka.producer.KafKaProducer;
import com.zhuwei.kafka.producer.NewProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.text.SimpleDateFormat;
import java.util.*;

/**
 * 功能；测试kafka发送消息
 * @author zhuiwei
 * @version 2019/05/14
 */
@RestController
public class SendMsgController {

    @Autowired
    private KafKaProducer producer;

    private Gson gson = new Gson();

    SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    /**
     * 发送消息
     */
    @GetMapping("/send")
    public void sendSimple(String msg){
        Map<String,String> map=new HashMap<>();
        map.put("msg",msg);
        map.put("data",format.format(new Date()));
        String beanJson=gson.toJson(map);
        producer.sendMessage(Topic.FIRST, beanJson);
    }

    /**
     * 向组发送消息
     */
    @GetMapping("/sendToPartition")
    public void sendGroup(String msg){
        List<String> list=new ArrayList<>();
        Map<String,String> map1=new HashMap<>();
        Map<String,String> map2=new HashMap<>();
        Map<String,String> map3=new HashMap<>();
        map1.put("msg",msg+System.currentTimeMillis()+1);
        map2.put("msg",msg+System.currentTimeMillis()+2);
        map3.put("msg",msg+System.currentTimeMillis()+3);
        String beanJson1=gson.toJson(map1);
        String beanJson2=gson.toJson(map2);
        String beanJson3=gson.toJson(map3);
        list.add(beanJson1);
        list.add(beanJson2);
        list.add(beanJson3);
        producer.sendMessage(Topic.GROUP, list,3);
    }
}
